package com.zte.claa.inficombo.csa.app.service.rptparse.ext.rawtransfer;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import com.zte.claa.inficombo.csa.app.config.RawTransferCommProps;
import com.zte.claa.inficombo.csa.app.dao.mysql.impl.ProjCustAttrMapper;
import com.zte.claa.inficombo.csa.app.model.proj.ProjCustAttrBean;
import com.zte.claa.inficombo.csa.app.service.rptparse.ParserContainer;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.ExtRptDataParser;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.RptRawData;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.comm.CommParamProps;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.comm.CommService;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.comm.CommServiceImpl;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.comm.mina.msgbean.MinaCommRequest;
import com.zte.claa.inficombo.csa.app.util.SecurityUtil;

/**
 * 通用原始数据透传服务实现类。
 * 
 * @author 0283000115
 */
@Service("DefaultRawTransferService")
@Lazy(false)
public class DefaultRawTransferService implements InitializingBean, DisposableBean, ExtRptDataParser {	
	public static final String RAW_DEST_SVRIP = "rawdatadestsvrip";
	public static final String RAW_DEST_SVRPORT = "rawdatadestsvrport";
	// 缓存队列大小
	private static final int QUEUE_MAX_SIZE = 1000;
	// 日志记录
	private final Logger logger = LoggerFactory.getLogger(DefaultRawTransferService.class);
	
	@Autowired
	private ParserContainer parserContainer;
	
	@Autowired
	private RawTransferCommProps rawCommProps;
	
	@Autowired
	private ProjCustAttrMapper custAttrMapper;
	
	@Value("${csa.updata.3rd.subparser.raw.devtps}")
	private String need3rdParseRawTransDevtps = "";
	
	// 内部分发队列...
	private Thread[] dataDealears = null;
	private LinkedBlockingQueue<RptRawData> queue = null;
	private boolean isWorkOn = true;
	
	// 透传目标地址mina通讯服务: projectid -- CommService 
	private Map<String, CommService> destSvrMap = null;
	
	/**
	 * 服务初始化.
	 */
	@Override
	public void afterPropertiesSet() throws Exception {
		this.register4Devtypes();
		this.initCommService();
		// 共用一个缓冲队列.
		this.queue = new LinkedBlockingQueue<RptRawData>(QUEUE_MAX_SIZE);
		this.isWorkOn = true;
		this.dataDealears = new Thread[this.rawCommProps.getTotaldealernum()];
		for(int i=0; i<dataDealears.length; i++){
			dataDealears[i] = new Thread(new Runnable() {
				@Override
				public void run() {
					dealRptRawData();
				}
			});
			dataDealears[i].setName("DefaultRawTransferService-dispatcher-"+i);
			dataDealears[i].start();
		}		
		logger.info("DefaultRawTransferService is inited ok, dealernum = "+this.rawCommProps.getTotaldealernum());
	}
	
	public String getNeed3rdParseRawTransDevtps() {
		return need3rdParseRawTransDevtps;
	}

	public void setNeed3rdParseRawTransDevtps(String need3rdParseRawTransDevtps) {
		this.need3rdParseRawTransDevtps = need3rdParseRawTransDevtps;
	}

	/**
	 * 解析配置的设备类型.
	 * @throws Exception
	 */
	private void register4Devtypes() throws Exception {
		// 根据配置文件配置，逐个设备类型分别注册... 
		if(need3rdParseRawTransDevtps == null || need3rdParseRawTransDevtps.trim().equals("")){
			logger.info("There is no raw transfer devtype config.");
			return;
		}
		String[] dtpArr = this.need3rdParseRawTransDevtps.trim().split(",");
		for(String dtp : dtpArr){
			if(dtp == null || dtp.trim().equals("")){
				continue;
			}
			this.parserContainer.setExtRptDataDealer(dtp.trim(), this);
			logger.info("setExtRptDataDealer for devtp ok: " + dtp);
		}		
	}
	
	/**
	 * 初始化底层通讯服务.
	 * 创建commsvr对象，并不实际建立连接.
	 * 
	 * @throws Exception
	 */
	private void initCommService() throws Exception {
		List<ProjCustAttrBean> attrList = this.custAttrMapper.queryAllCustAttrBeans();
		if(attrList == null || attrList.size() == 0){
			return;
		}		
		Map<String, CommParamProps> projCommParaMap = new HashMap<String, CommParamProps>();
		for(ProjCustAttrBean pcab : attrList){
			String projectId = pcab.getProjectid();
			if(projectId == null || projectId.trim().equals("")){
				continue;
			}
			String attrName = pcab.getAttrname();
			if(attrName == null || attrName.trim().equals("")){
				continue;
			}
			attrName = attrName.trim();
			if(RAW_DEST_SVRIP.equalsIgnoreCase(attrName)){
				String svrIp = pcab.getAttrvalue();
				if(svrIp == null || svrIp.trim().equals("")){
					continue;
				}					
				CommParamProps cpp = projCommParaMap.get(projectId);
				if(cpp == null){
					cpp = this.transferCommProps();
					projCommParaMap.put(projectId, cpp);
				}				
				cpp.setSvrip(svrIp.trim());
			}
			else if(RAW_DEST_SVRPORT.equalsIgnoreCase(attrName)){
				String svrPort = pcab.getAttrvalue();
				if(svrPort == null || svrPort.trim().equals("")){
					continue;
				}	
				int port = 0;
				try{
					port = Integer.parseInt(svrPort.trim());
				}catch(Exception e){
					logger.warn("Dest svr port is invalid: "+svrPort);
					continue;
				}
				CommParamProps cpp = projCommParaMap.get(projectId);
				if(cpp == null){
					cpp = this.transferCommProps();
					projCommParaMap.put(projectId, cpp);
				}				
				cpp.setPort(port);
			}
		}
		this.destSvrMap = new HashMap<String, CommService>();
		Iterator<String> it = projCommParaMap.keySet().iterator();
		while(it.hasNext()){
			String projId = it.next();
			CommParamProps cpprops = projCommParaMap.get(projId);
			if(cpprops == null){
				continue;
			}
			// 过滤掉无效值.
			if(cpprops.getSvrip() == null || cpprops.getSvrip().equals("")){
				continue;
			}
			if(cpprops.getPort() == 0){
				continue;
			}
			this.destSvrMap.put(projId, new CommServiceImpl(cpprops));
		}
	}
	
	/**
	 * 添加新的目标地址
	 * 
	 * @param projectId
	 * @param destSvrIp
	 * @param destSvrPort
	 * @throws Exception
	 */
	public void addNewDestSvr(String projectId, String destSvrIp, int destSvrPort) throws Exception {
		CommParamProps cpp = this.transferCommProps();
		cpp.setSvrip(destSvrIp);
		cpp.setPort(destSvrPort);
		this.destSvrMap.put(projectId, new CommServiceImpl(cpp));
	}
	
	/**
	 * 删除目标服务器.
	 * 后台维护入口调用.
	 * 
	 * @param projectId
	 * @throws Exception
	 */
	public void removeDestSvr(String projectId) throws Exception {
		if(this.destSvrMap != null){
			CommService commSvr = this.destSvrMap.remove(projectId);
			if(commSvr != null){
				commSvr.closeConnectToDestSvr();
			}
			logger.info("Remove dest svr commservice ok: "+projectId);
		}
	}
	
	/**
	 * 退出销毁资源.
	 */
	@Override
	public void destroy() throws Exception {
		this.isWorkOn = false;
		if(this.queue != null){
			this.queue.clear();
			for(int i=0; i<this.rawCommProps.getTotaldealernum(); i++){
				this.queue.offer(new RptRawData(RptRawData.INVALID_DEVEUI));
			}
		}
		
		// 关闭连接
		if(this.destSvrMap != null){
			Iterator<String> it = this.destSvrMap.keySet().iterator();
			while(it.hasNext()){
				String projId = it.next();
				CommService commSvr = this.destSvrMap.get(projId);
				if(commSvr != null){
					commSvr.closeConnectToDestSvr();
				}
			}
		}
	}
	
	/**
	 * 
	 * @return
	 */
	private CommParamProps transferCommProps() {
		CommParamProps cpp = new CommParamProps();
		cpp.setBufsize(this.rawCommProps.getBufsize());
		cpp.setCharsetname(this.rawCommProps.getCharsetname());
		cpp.setCmdsendtimeoutsecond(this.rawCommProps.getCmdsendtimeoutsecond());
		cpp.setCmdtimeoutsecond(this.rawCommProps.getCmdtimeoutsecond());
		cpp.setConnwaitsecond(this.rawCommProps.getConnwaitsecond());
		cpp.setSessionidlesecond(this.rawCommProps.getSessionidlesecond());
		return cpp;
	}

	/**
	 * 分发采集器上报的原始码流...
	 */
	@Override
	public void processRptData(RptRawData data) throws Exception {
		// 添加到linkedqueue中，等待解析处理...
		boolean b = this.queue.offer(data);
		if(!b){
			logger.warn("[*!!*]Add data to queue failed: "+data.toString());
		}
	}
	
	/**
	 * 分配处理数据.
	 */
	private void dealRptRawData() {
		while (this.isWorkOn) {
			CommService commSvr = null;
			try {
				// 发送原始码流到Dest服务器进行数据转换...
				RptRawData rawData = this.queue.poll(120, TimeUnit.SECONDS);
				if(rawData == null){
					logger.debug("No raw data available now!");
					continue;
				} 
				if(RptRawData.INVALID_DEVEUI.equalsIgnoreCase(rawData.getDeveui())){
					continue;
				}
				String projectId = rawData.getProjectid();
				commSvr = this.destSvrMap.get(projectId);
				if(commSvr == null){
					logger.info("There is no commservice for project: "+projectId);
					continue;
				}				
				byte[] bs = rawData.getPayload();
				bs = (bs == null) ? new byte[0] : bs;
				logger.info("Begin submit raw data: "+rawData+", size="+bs.length+", data="+SecurityUtil.parseByte2HexStr(bs));
				MinaCommRequest req = new MinaCommRequest();
				req.setRawData(rawData);
				commSvr.execCommandIngoreRsp(req);
				try{
					Thread.sleep(2);
				}catch(Throwable ignore){}
			} catch (Throwable t) {
				// 这里做一个保护，如果发送失败，则尝试关闭到目标服务器的链路。注意不能赋值为null.
				try {
					commSvr.closeConnectToDestSvr();
				} catch (Exception e) {
					logger.error(e.getMessage(), e);
				}
				logger.error("dealRptRawData failed: " + t.getMessage(), t);
			}
		}
		logger.warn("Dealer thread exit: " + Thread.currentThread().getName());
	}
}
